package com.project.mq.mqserver.datacenter;

import com.project.mq.common.BinaryTool;
import com.project.mq.common.MqException;
import com.project.mq.mqserver.core.MSGQueue;
import com.project.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

/**
 * 通过这个类，来针对硬盘上的消息进行管理。
 */
public class MessageFileManager {
    // 此处定义一个内部类,表示消息的统计信息
    //优先考虑使用static静态内部类
    static public class Stat {
        //此处直接定义成public,就不再搞get/set方法了
        //对于这样的简单的类，就直接使用成员，类似于C的结构体了。
        // 这意味着在设计简单类时，可以采用类似C语言中结构体的方式来组织数据成员，
        // 而不需要过多地封装方法或行为。这样做的好处是简洁明了，易于理解和维护。
        public int totalCount;//总消息数
        public int validCount;//有效消息树
    }

    public void init() {
        // 当前这里不需要做任何工作.以备后续扩展
    }
    //约定消息文件所在的目录名和文件名
    // 队列目录
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }
    // 队列数据文件
    // 这个文件来存储队列的真实数据
    //二进制使用txt有点不合适,使用.bin / .dat更合适,此处忽略这个细节
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }
    // 队列统计文件
    // 这个文件用来存储队列中的统计信息.
    // 包含一行，两个列使用\t分割，分别是总数据，和无效数据.
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }
    // 从统计文件中读取结果
    private Stat readStat(String queueName) {
        //由于当前的消息统计文件是文本文件，可以直接使用Scanner 来读取文件内容
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    // 向统计文件中写入结果
    private void writeStat(String queueName, Stat stat) {
        //PrintWriter写文件
        // OutputStream 打开文件，默认情况下，会直接把原文件清空。此时相当于新的数据覆盖了旧的。,
        // 不想覆盖在new FileOutputStream(getQueueStatPath(queueName),ture)加ture
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 创建目录指定队列的目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }

        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }

        // 3. 创建队列统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }

        // 4. 给队列统计文件写入初始数据
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }
    //实现删除队列⽬录文件
    //如果队列需要删除, 则队列对应的⽬录/⽂件也需要删除.
    public void destroyQueueFiles(String queueName) throws IOException {
        // 1. 先删除目录中的文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();

        // 2. 再删除目录。delete 要求必须是空目录才能删除。
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();

        if (!ok1 || !ok2 || !ok3) {
            //任意一个删除失败,则失败
            throw new IOException("删除队列目录失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }
    //检查队列⽂件是否存在
    //判定该队列的消息⽂件和统计⽂件是否存在. ⼀旦出现缺失, 则不能进⾏后续⼯作.
    //比如后续有生产者给broker server 生产消息了， 这个消息就可能需要记录到文件上(取决于消息是否要持久化)
    private boolean checkFilesExists(String queueName) {
        File queueData = new File(getQueueDataPath(queueName));
        if (!queueData.exists()) {
            return false;
        }
        File queueStat = new File(getQueueStatPath(queueName));
        if (!queueStat.exists()) {
            return false;
        }
        return true;
    }
    //实现写⼊消息⽂件
    //这个方法用来把一个新的消息，放到队列对应的文件中。
    // queue 表示要把消息写入的队列，message 则是要写的消息。
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //0.检查一下当前要写入的队列对应的文件是否存在
        if (!checkFilesExists(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列匹配的文件不存在! queueName=" + queue.getName());
        }
        // 1. 先把 message 转成二进制
        byte[] messageBinary = BinaryTool.toBytes(message);
        // 此处的锁对象以队列为维度. 不同队列之间不涉及锁冲突.
        synchronized (queue) {
            // 2. 先获取到文件总长度,用来计算offsetbeg和offsetend
            //把新的 Message 数据，写入到队列数据文件的末尾。
            // 此时 Message 对象的 offsetBeg，就是当前文件长度 + 4；
            // offsetEnd 就是当前文件长度 + 4 + message 自身长度。
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            // 3. 写入消息数据文件,此处应该是追加写的方式new FileOutputStream(queueDataFile, true)
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 先写入消息长度
                    dataOutputStream.writeInt(messageBinary.length);
                    // 再写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }

            // 4. 写入消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }
    //实现删除消息
    //此处的删除只是 "逻辑删除", 即把 Message 类中的 isValid 字段设置为 0.
    //这样删除速度⽐较快. 实际的彻底删除, 则通过我们⾃⼰实现的 GC 来解决.
    // 把文件上的对应消息给删除掉。(标记成无效)
    //此处这个参数中的 message 对象，必须得包含有效的 offsetBeg 和 offsetEnd。
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 先从文件中读取出 Message 的数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);

                // 2. 转成 Message 对象,反序列化
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);

                // 3. 设置成无效
                //此处不需要给参数的这个 message 的 isValid 设为 0，
                // 因为这个参数代表的是内存中管理的 Message 对象，而这个对象马上也要被从内存中销毁了。
                diskMessage.setIsValid((byte) 0x0);

                // 4. 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                //虽然上面已经 seek 过了，但是上面 seek 完了之后，进行了读操作，这一读，就导致，文件光标往后移动，
                // 移动到下一个消息的位置了,因此要想接下来的写入,能够刚好写回到之前的位置,就需要重新调整文件光标
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
                // 通过上述这通折腾，对于文件来说，只是有一个字节发生改变而已了~
            }

            // 更新统计文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }
    //实现消息加载
    //把消息内容从⽂件加载到内存中. 这个功能在服务器重启进行调用, 放到一个链表中.
    //这里使用一个LinkedList，主要目的是为了后续进行头删操作。
    // 从消息数据文件中读取出所有消息
    //这个方法的参数，只是一个 queueName 而不是 MSGQueue 对象。因为这个方法不需要加锁，只使用 queueName 就够了。
    // 由于该方法是在程序启动时调用，此时服务器还不能处理请求呢~~不涉及多线程操作文件。
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws MqException, IOException, ClassNotFoundException {
        // 记录当前读到的数据在文件的 offset 光标
        long currentOffset = 0;
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                //一个文件包含了很多消息,此处势必要循环读取
                while (true) {
                    //1.读取当前消息的长度,此处可能会读到文件末尾
                    // 读到文件末尾，会触发 EOFException
                    int messageSize = dataInputStream.readInt();
                    //2.按照这个长度读取内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    if (message.getIsValid() != 0x1) {
                        // 被删除的无效数据，直接跳过.不要忘记更新 currentOffset
                        currentOffset += 4 + messageSize;
                        continue;
                    }
                    // 计算该 message 的 offset
                    //有效数据，则需要把这个Message对象加入到链表中，加入之前还需要填写offsetBeg和offsetEnd
                    //进行计算offset的时候，需要知道当前文件光标的位置的
                    //，由于当下使用的DataInputStream并不方便直接获取到文光标
                    //因此就需要手动计算下文件光标。
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    // 每个消息，开头4个字节保存的是消息的长度.接下来 [offsetBeg, offsetEnd) 是消息体
                    currentOffset += 4 + messageSize;
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 数据读取完毕，循环正常退出!
                //这个catch并非真是处理"异常"，而是处理"正常"的业务逻辑。
                // 文件读到末尾，会被readInt抛出该异常。这个catch语句中也不需要做啥特殊的事情。
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
        return messages;
    }
    //实现垃圾回收(GC)
    // 检查是否要对文件进行GC操作
    public boolean checkGC(String queueName) {
        // 判定是否要 GC，是根据总消息数和有效消息数，这两个值都是在消息统计文件中的。
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount <= 0.5) {
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }
    // 真正执行GC操作,复制算法完成
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行gc的时候，是针对消息数据文件进行大洗牌。在这个过程中，其他线程不能针对该队列的消息文件做任何修改。
        synchronized (queue) {
            //由于gc操作可能比较耗时,此处统计一下消耗时间
            long gcBeg = System.currentTimeMillis();
            // 1. 创建一个新的文件，名字为“queue_data_new.txt”
            File queueDataNew = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNew.exists()) {
                //正常情况下，这个文件不应该存在。如果存在，就是意外~~说明上次 gc了一半，程序意外崩溃了。
                throw new MqException("[MessageFileManager] gc时发现列队数据文件已经存在!queueName=" + queue.getName());
            }
            boolean ok = queueDataNew.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败!queueDataNew=" + queueDataNew.getAbsolutePath());
            }

            // 2. 遍历旧文件，读取出每个消息（只保留有效消息）调用上述方法即可
            List<Message> messageList = loadAllMessageFromQueue(queue.getName());

            // 3. 将所有有效消息写入新的文件中
            try (OutputStream outputStream = new FileOutputStream(queueDataNew)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messageList) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }

            // 4. 删除queue_data.txt，后将queue_data_new.txt重命名为queue_data.txt
            File queueDataOld = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOld.delete();
            if (!ok) {
                throw new IOException("删除文件失败!queueDataOld=" + queueDataOld.getAbsolutePath());
            }
            //queue_data_new.txt重命名为queue_data.txt
            ok = queueDataNew.renameTo(queueDataOld);
            if (!ok) {
                throw new IOException("文件重命名失败!queueDataOld=" + queueDataOld.getAbsolutePath() + ", queueDataNew=" + queueDataNew.getAbsolutePath());
            }

            // 5. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.validCount = messageList.size();
            stat.totalCount = messageList.size();
            writeStat(queue.getName(), stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc执行完!queueName=" + queue.getName() + ", time=" + (gcEnd - gcBeg) + "ms");
        }
    }
}
